package com.cetc.sdp.kmga.cs.stream;

import com.cetc.sdp.kmga.cs.device.DeviceExpCheck;
import com.cetc.sdp.kmga.cs.device.DeviceExpCode;
import com.cetc.sdp.kmga.cs.device.DeviceExpEvent;
import com.cetc.sdp.kmga.cs.device.StateLog;
import com.cetc.sdp.kmga.cs.jdbc.DeviceDAO;
import com.cetc.sdp.kmga.cs.util.Tool;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 设备状态日志处理（基于Kafka High Level API）
 */
public final class DeviceStateLogConsumer {
    /**
     * kafka消费者不是线程安全的
     */
    private final KafkaConsumer<String, String> consumer;

    private DeviceExpCheck deviceExpCheck;

    private ExecutorService executorService;

    private int offlineCount;

    private static final String STATE_STREAM_PROP_PATH = "StateStream.properties";

    private static final Logger L = LoggerFactory.getLogger(DeviceStateLogConsumer.class);

    /**
     * 设备在线状态缓存
     */
    private Map<String, Integer> deviceStateCache = new ConcurrentHashMap<>();

    public DeviceStateLogConsumer(String bootstrapServers, String groupId, String topic, int offlineCount) {
        L.info("初始化kafka消费者队列");
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        props.put("group.id", groupId);
        // 关闭自动提交
        props.put("enable.auto.commit", "false");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList(topic));
        deviceExpCheck = new DeviceExpCheck(DeviceDAO.findAllDeviceExpCode());
        this.offlineCount = offlineCount;

        executorService = Executors.newFixedThreadPool(5);

        List<Tuple2<String,Integer>> devices = DeviceDAO.findAllDevice();
        devices.forEach(dev -> deviceStateCache.put(dev._1(), offlineCount));
    }

    public void consumer(int interval) {
        L.info("开始消费");
        DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

        Map<String, StateLog> devSet = new HashMap<>();
        Map<String, List<DeviceExpEvent>> devExpEventMap = new HashMap<>();
        int offlineExpCode = DeviceDAO.getOfflineExpCode();
        while (true) {
            try {
                ConsumerRecords<String, String> records = consumer.poll(interval * 1000);
                String time = LocalDateTime.now().format(dateTimeFormatter);
                List<StateLog> data = new ArrayList<>(records.count());
                records.forEach(record -> {
                    StateLog log = StateStream.parseDevStatLog(record.value().toString());
                    String devID = log.getDevId();
                    if (devID != null) {
                        data.add(log);
                        devSet.put(devID, log);
                        //异常检测
                        List<DeviceExpEvent> expEvents = deviceExpCheck.check(log);
                        devExpEventMap.put(devID, expEvents);
                    }
                });

                //离线检查
                DeviceExpCode expCode = deviceExpCheck.getOfflineExpCode();
                for (Map.Entry<String, Integer> devEntry : deviceStateCache.entrySet()) {
                    String dev = devEntry.getKey();
                    Integer count = devEntry.getValue();
                    StateLog log = devSet.get(dev);
                    if (log != null) {
                        deviceStateCache.put(dev, this.offlineCount);
                    } else {
                        int c = count - 1;
                        if (c == 0) {
                            devExpEventMap.put(dev, Arrays.asList(makeOfflineExpEvent(dev, time, expCode)));
                            deviceStateCache.put(dev, 0);
                        } else if (c > 0) {
                            deviceStateCache.put(dev, c);
                        } else {
                            devExpEventMap.put(dev, Arrays.asList(makeOfflineExpEvent(dev, time, expCode)));
                        }
                    }
                }

                Map<String, List<DeviceExpEvent>> activeExpEvent = DeviceDAO.findAllActiveExpEvent();
                for(Map.Entry<String,  List<DeviceExpEvent>> x :  devExpEventMap.entrySet()) {
                    String dev = x.getKey();
                    List<DeviceExpEvent> expEvents = x.getValue();
                    if (expEvents.isEmpty()) {
                        //无异常
                        List<DeviceExpEvent> tmp = activeExpEvent.get(dev);
                        if (tmp != null && !tmp.isEmpty()) {
                            //添加异常结束时间
                            for (DeviceExpEvent expEvent : tmp) {
                                expEvent.setEndTime(time);
                                DeviceDAO.updateDeviceExpEvent(expEvent);
                            }
                        }
                    } else {
                        //有异常
                        List<DeviceExpEvent> tmp = activeExpEvent.get(dev);
                        if (tmp == null || tmp.isEmpty()) {
                            //添加新异常
                            for (DeviceExpEvent expEvent : expEvents) {
                                DeviceDAO.saveDeviceExpEvent(expEvent);
                            }
                        } else {
                            for (DeviceExpEvent expEvent : expEvents) {
                                boolean flag = true;
                                for (DeviceExpEvent oldExpEvent : tmp) {
                                    if (oldExpEvent.getExcpCode() == expEvent.getExcpCode()) {
                                        flag = false;
                                        break;
                                    }
                                }
                                if (flag) {
                                    //当设备离线时，该设备的其余告警事件结束
                                    if (expEvent.getExcpCode() == offlineExpCode) {
                                        tmp.forEach(event -> {
                                            event.setEndTime(time);
                                            DeviceDAO.updateDeviceExpEvent(event);
                                        });
                                    }
                                    DeviceDAO.saveDeviceExpEvent(expEvent);
                                }
                            }
                        }
                    }
                }

                executorService.submit(() -> {
                   //save log
                    DeviceDAO.saveDeviceStateLog(data, time);
                });
                devExpEventMap.clear();
                devSet.clear();
                consumer.commitSync();
            } catch (Exception e) {
                L.error("消费出错");
                L.error(e.getMessage(), e.getCause());
            }
        }
    }

    private DeviceExpEvent makeOfflineExpEvent(String devId, String time, DeviceExpCode expCode) {
        DeviceExpEvent expEvent = new DeviceExpEvent();
        expEvent.setStartTime(time);
        expEvent.setExcpCode(expCode.getExcpCode());
        expEvent.setDeviceId(devId);
        return expEvent;
    }

    public void shutdown() {
        try {
            if (consumer != null) {
                consumer.close();
            }
            if (executorService != null) {
                executorService.shutdown();
            }
            if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {
                System.out.println("Timeout");
            }
        } catch (InterruptedException ignored) {
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) {
        Properties prop = Tool.loadProperties(STATE_STREAM_PROP_PATH);
        String bootstrapServers = prop.getProperty("kafka.bootstrap.server");
        String groupId = prop.getProperty("kafka.group.id");
        String topic = prop.getProperty("kafka.stream.topic");
        String offlineCountStr = prop.getProperty("max.offline.count", "3");

        int interval = Integer.parseInt(args[0]);
        int offlineCount = Integer.parseInt(offlineCountStr);

        DeviceStateLogConsumer consumer = new DeviceStateLogConsumer(bootstrapServers, groupId, topic, offlineCount);
        consumer.consumer(interval);
    }
}
